package cgl.hadoopsensorgrid.sensorgrid.eval;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Hashtable;
import java.util.LinkedList;
import java.util.Properties;
import java.util.Queue;
import java.util.Random;
import java.util.StringTokenizer;
import java.lang.InterruptedException;

import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.Text;

import cgl.sensorgrid.common.PropertyFile;
import cgl.sensorgrid.sopac.gps.threadpool.ThreadPool;
import cgl.sensorgrid.sopac.gps.filters.UtilSet;

import cgl.hpsearch.common.Sys;

/**
 * 
 * posEval class periodically builds models for each station's POS data and
 * evaluates the data. It extends the Mapper class.
 * 
 * @author Chao Sun
 * 
 */
public class posEval extends Mapper<Object, Text, Text, BytesWritable> {
	private Properties properties;
	private String evalTempPath;
	private String siteName;
	private String modelFilePrefix;
	private String modelInputPrefix;
	private String modelParentDir;
	private String evalParentDir;
	private String siteModelDir;
	private String siteEvalDir;
	private long dayMilli = 86400 * 1000;
	private long lastRemodelTimeMilli = 0;
	private long lastEvalTimeMilli = 0;
	private long modelInputStartMilli = 0;
	private boolean isModelBuilt;
	private boolean lastModelUsableValue;
	private LinkedList stateChangeTimes;
	private long remodelPeriodMilli = -1;
	private long evalPeriodMilli = -1;
	private long modelDataTimeMilli = -1;
	private String rdahmmTrainShellPath = null;
	private String rdahmmEvalShellPath = null;
	private String plotShellPath = null;
	private Calendar calNow = Calendar.getInstance();
	private ThreadPool modelThreadPool;

	/**
	 * Load properties from a given file.
	 */
	private void loadProperties() {
		try {
			properties = PropertyFile.loadProperties("hadoopsensorgrid.properties");
			evalTempPath = properties.getProperty("hadoop.eval.outputbase");
			evalParentDir = properties.getProperty("rdahmm.eval.directory");
			modelParentDir = properties.getProperty("rdahmm.data.directory");
			rdahmmTrainShellPath = properties
					.getProperty("rdahmm.trainshell.directory");
			rdahmmEvalShellPath = properties
					.getProperty("rdahmm.evalshell.directory");
			plotShellPath = properties.getProperty("rdahmm.plotshell.directory");
			remodelPeriodMilli = Long.parseLong(properties
					.getProperty("rdahmm.remodel.days"))
					* dayMilli;
			/*
			 * evalPeriodMilli = Long.parseLong(properties
			 * .getProperty("rdahmm.eval.minutes")) * 60000; modelDataTimeMilli =
			 * Long.parseLong(properties .getProperty("rdahmm.modeldata.hours")) * 60
			 * * 60000;
			 */
			evalPeriodMilli = 40000;
			modelDataTimeMilli = 60000;
		} catch (Exception e) {
			e.printStackTrace();
		}

		if (properties == null || evalParentDir == null || modelParentDir == null
				|| rdahmmTrainShellPath == null || rdahmmEvalShellPath == null
				|| plotShellPath == null || remodelPeriodMilli == -1
				|| evalPeriodMilli == -1 || modelDataTimeMilli == -1) {
			System.err.println("Fatal Error: failed to load properties!");
			System.exit(-1);
		}
	}

	/**
	 * Check to see if a model already exists and is up to date.
	 */
	private void checkModelExists() {
		File fModel = new File(siteModelDir + modelFilePrefix + ".Q");
		isModelBuilt = fModel.exists();
		if (isModelBuilt) {
			lastRemodelTimeMilli = fModel.lastModified();
			if (remodelPeriodMilli > 0
					&& System.currentTimeMillis() - lastRemodelTimeMilli >= remodelPeriodMilli) {
				isModelBuilt = false;
			}
		}
	}

	/**
	 * Periodically check if it is time to evaluate the data or rebuild the model.
	 * 
	 * @throws InterruptedException
	 */
	private void run() throws InterruptedException {
		while (true) {
			// if no model is built yet, check if it is time to build one
			if (!isModelBuilt) {
				try {
					if (modelInputStartMilli == 0) {
						File timeFile = new File(siteModelDir + siteName
								+ "_input_start_milli");
						while (!timeFile.exists()) {
							Thread.sleep(5000);
						}
						FileReader timeReader = new FileReader(siteModelDir + siteName
								+ "_input_start_milli");
						BufferedReader timeBuffer = new BufferedReader(timeReader);
						modelInputStartMilli = Long.parseLong(timeBuffer.readLine());
					}
					if (System.currentTimeMillis() - modelInputStartMilli >= modelDataTimeMilli) {
						buildModel();
						// *************************** Testing ***************************
						/*Calendar now = Calendar.getInstance();
						System.out.println(getDateString(now, true) + ": Build Model: "
								+ siteName);
						isModelBuilt = true;*/
						// *************************** Testing ***************************
						modelInputStartMilli = 0;
					}
				} catch (Exception e) {
					e.printStackTrace();
				}
			}

			// if a new model has not been built, check if there is an old one to use
			boolean modelUsable = isModelBuilt;

			if (modelUsable) {
				System.out.println("Model usable: " + siteName);
			}

			if (!modelUsable) {
				File modelFile = new File(siteModelDir + modelFilePrefix + ".Q");
				if (modelFile.exists()) {
					long timeSinceLastModel = System.currentTimeMillis()
							- modelFile.lastModified();
					modelUsable = timeSinceLastModel < (remodelPeriodMilli
							+ modelDataTimeMilli + evalPeriodMilli);
					if (modelUsable && modelUsable != lastModelUsableValue) {
						System.out.println("Old model usable: " + siteName);
						log("Old model usable for site" + siteName);
					}
				}
			}

			// if there is a model to use, check if it is time to evaluate
			lastModelUsableValue = modelUsable;
			if (modelUsable) {
				try {
					if (lastEvalTimeMilli == 0) {
						System.out.println("Starting timing for the first evaluation: "
								+ siteName);
						lastEvalTimeMilli = System.currentTimeMillis();
					}
					if (System.currentTimeMillis() - lastEvalTimeMilli >= evalPeriodMilli) {
						System.out.println("Time to evaluate: " + siteName);
						evaluate();
						// *************************** Testing ***************************
						/*Calendar now = Calendar.getInstance();
						System.out.println(getDateString(now, true) + ": Evaluate: "
								+ siteName);
						lastEvalTimeMilli = System.currentTimeMillis();*/
						// *************************** Testing ***************************
						if (remodelPeriodMilli > 0
								&& System.currentTimeMillis() - lastRemodelTimeMilli >= remodelPeriodMilli) {
							isModelBuilt = false;
						}
					}
				} catch (Exception e) {
					e.printStackTrace();
				}
			}

			// sleep for a while
			Thread.sleep(10000);
		}
	}

	/**
	 * Build a model.
	 * 
	 * @throws InterruptedException
	 */
	private void buildModel() throws InterruptedException {
		String modelInputFileName = siteModelDir + modelFilePrefix + ".xyz";
		File modelInputFile = new File(modelInputFileName);
		if (modelInputFile.exists()) {
			File evalTempDir = new File(evalTempPath);
			// wait until the input file is not written into
			while (evalTempDir.exists()) {
				Thread.sleep(1000);
			}
			// rename the input file. if there is an old one, delete it first
			String renamedModelInput = siteModelDir + modelInputPrefix + ".xyz";
			File renamedModelFile = new File(renamedModelInput);
			if (renamedModelFile.exists()) {
				String rmCommand = "rm " + modelInputFileName;
				exec(rmCommand);
			}
			String mvCommand = "mv " + modelInputFileName + " " + renamedModelInput;
			exec(mvCommand);
			// model in another thread
			ModelThread mt = new ModelThread(rdahmmTrainShellPath, renamedModelInput,
					modelThreadPool);
			mt.start();
			lastRemodelTimeMilli = System.currentTimeMillis();
			isModelBuilt = true;
		} else {
			System.out.println("Input file does not exist: " + modelInputFileName);
		}
	}

	/**
	 * Evaluate.
	 */
	private void evaluate() {
		String input = siteEvalDir + siteName + ".xyz";
		String command = rdahmmEvalShellPath + " " + input + " 5 "
				+ modelInputPrefix + " " + siteModelDir;
		String out = exec(command);
		if (out.length() > 0) {
			log("Results for RDAHMM evaluation for " + siteName + ": " + out);
		} else {
			log("Sucessfully done RDAHMM evaluation for " + siteName);
		}
		
		// run plot script.
		// *************************** Testing ***************************
		/*Calendar now = Calendar.getInstance();
		System.out.println(getDateString(now, true) + ": Plot: "
				+ siteName);*/
		// *************************** Testing ***************************
		String xyzFilePath = input + ".xyz";
		String QFilePath = siteEvalDir + siteName + ".Q";
		command = plotShellPath + " " + xyzFilePath + " " + QFilePath;
		out = exec(command);
		if (out.length() > 0) {
			log("Results for plotting for " + siteName + ": " + out);
		} else {
			log("Sucessfully plotted for " + siteName);
		}
		lastEvalTimeMilli = System.currentTimeMillis();

		checkStateChange(input, QFilePath);
		recordStateChangeSummary();

		// delete eval file
		File inputFile = new File(input);
		inputFile.delete();
		System.out.println("Delete evaluation input file for site: " + siteName);
	}

	private void checkStateChange(String inputFilePath, String qFilePath) {
		try {
			String line1, line2, inputChange, input2;
			BufferedReader brQ = new BufferedReader(new FileReader(qFilePath));
			line1 = brQ.readLine();
			line2 = brQ.readLine();
			BufferedReader brInput = new BufferedReader(new FileReader(inputFilePath));
			input2 = brInput.readLine();
			input2 = brInput.readLine();

			inputChange = null;
			// looking for state changes
			while (line2 != null) {
				if (!line1.equals(line2)) {
					inputChange = input2;
					// **************** Need modify this block later ****************
					// publishData(makeResultMsg(line1, line2, input2).getBytes());
					// **************** Need modify this block later ****************
				}
				line1 = line2;
				line2 = brQ.readLine();
				input2 = brInput.readLine();
			}
			brQ.close();
			brInput.close();

			// if there a state change, record the time of it
			if (inputChange != null) {
				System.out.println("Line1 in checkStateChange:" + inputChange);
				int tab1, tab2, tab3;
				tab1 = inputChange.indexOf('\t');
				tab2 = inputChange.indexOf('\t', tab1 + 1);
				tab3 = inputChange.indexOf('\t', tab2 + 1);
				Calendar changeTime = Calendar.getInstance();
				UtilSet.setDateByString(changeTime, inputChange.substring(tab1 + 1,
						tab2));
				String timeStr = inputChange.substring(tab2 + 1, tab3);
				System.out.println("TimeStr in checkStateChange:" + timeStr);
				int colon1 = timeStr.indexOf(':');
				int colon2 = timeStr.indexOf(':', colon1 + 1);
				int posAMPM = timeStr.indexOf('M') - 1;
				changeTime.set(Calendar.HOUR, Integer.parseInt(timeStr.substring(0,
						colon1)));
				changeTime.set(Calendar.MINUTE, Integer.parseInt(timeStr.substring(
						colon1 + 1, colon2)));
				changeTime.set(Calendar.SECOND, Integer.parseInt(timeStr.substring(
						colon2 + 1, posAMPM)));
				if (timeStr.charAt(posAMPM) == 'A')
					changeTime.set(Calendar.AM_PM, Calendar.AM);
				else
					changeTime.set(Calendar.AM_PM, Calendar.PM);

				stateChangeTimes.add(changeTime);
				if (stateChangeTimes.size() > (86400000 / evalPeriodMilli))
					stateChangeTimes.remove();
				System.out.println("State change found for " + siteName + " at "
						+ getDateString(changeTime, true));
			} else {
				stateChangeTimes.add(null);
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	private String makeResultMsg(String oldStatus, String newStatus,
			String inputLine) {
		int tab1, tab2, tab3;
		tab1 = inputLine.indexOf('\t');
		tab2 = inputLine.indexOf('\t', tab1 + 1);
		tab3 = inputLine.indexOf('\t', tab2 + 1);

		String stationName = inputLine.substring(0, tab1).toLowerCase();
		Calendar changeTime = Calendar.getInstance();
		UtilSet.setDateByString(changeTime, inputLine.substring(tab1 + 1, tab2));
		String timeStr = inputLine.substring(tab2 + 1, tab3);
		System.out.println("TimeStr in checkStateChange:" + timeStr);
		int colon1 = timeStr.indexOf(':');
		int colon2 = timeStr.indexOf(':', colon1 + 1);
		int posAMPM = timeStr.indexOf('M') - 1;
		int hour = Integer.parseInt(timeStr.substring(0, colon1));
		if (timeStr.charAt(posAMPM) == 'P' && hour < 12)
			hour += 12;
		changeTime.set(Calendar.HOUR_OF_DAY, hour);
		changeTime.set(Calendar.MINUTE, Integer.parseInt(timeStr.substring(
				colon1 + 1, colon2)));
		changeTime.set(Calendar.SECOND, Integer.parseInt(timeStr.substring(
				colon2 + 1, posAMPM)));

		return stationName + ';' + UtilSet.getDateTimeString(changeTime) + ';'
				+ oldStatus + "to" + newStatus + "\nEOF\n";
	}

	/**
	 * Check if the station has changed its state recently within timeRange
	 */
	private boolean isChangedInRecent(long timeRange) {
		Calendar now = Calendar.getInstance();
		for (int i = stateChangeTimes.size() - 1; i >= 0; i--) {
			Calendar changeTime = (Calendar) stateChangeTimes.get(i);
			if (changeTime != null) {
				if (now.getTimeInMillis() - changeTime.getTimeInMillis() <= timeRange)
					return true;
				else
					return false;
			}
		}

		return false;
	}

	/**
	 * Execute a command
	 * 
	 * @param execStr
	 *          String command string
	 * @return String the information of execute result
	 */
	private String exec(String execStr) {
		Runtime runtime = Runtime.getRuntime();
		String outInfo = "";
		try {
			String osname = System.getProperty("os.name");
			String[] args = null;
			if (osname.equals("Linux")) {
				args = new String[] { "sh", "-c", execStr };
			} else if (osname.substring(0, 3).equals("Win")) {
				args = new String[] { "cmd", "/c", execStr };
			} else {
				System.exit(0);
			}

			Process proc = runtime.exec(args);
			InputStream out = proc.getInputStream();
			BufferedReader br1 = new BufferedReader(new InputStreamReader(out));
			while (br1.readLine() != null) {
			}
			out.close();

			InputStream in = proc.getErrorStream();
			BufferedReader br = new BufferedReader(new InputStreamReader(in));
			String line = "";
			while ((line = br.readLine()) != null) {
				outInfo = outInfo + line + "\n";
			}
			in.close();

			try {
				if (proc.waitFor() != 0) {
					System.err.println("exit value = " + proc.exitValue());
				}
			} catch (InterruptedException e) {
				System.err.print(e);
				e.printStackTrace();
			}

		} catch (IOException e) {
			System.out.println("exec error: " + e.getMessage());
			e.printStackTrace();
		}
		return outInfo;
	}

	/**
	 * Save the state change information of every station
	 */
	private void recordStateChangeSummary() {
		try {
			FileWriter fw = new FileWriter(evalParentDir + File.separator
					+ "state_change" + ".txt", false);
			fw.write(siteName.toLowerCase());
			if (isChangedInRecent(7200000)) {
				fw.write(" yes");
			} else {
				fw.write(" no");
			}
			if (isChangedInRecent(86400000)) {
				fw.write(" yes\n");
			} else {
				fw.write(" no\n");
			}
			fw.flush();
			fw.close();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * Convert current day time to string.
	 */
	public String getDateString(Calendar date, boolean timeWanted) {
		StringBuffer ret = new StringBuffer();
		ret.append(date.get(Calendar.YEAR)).append('-').append(
				date.get(Calendar.MONTH) + 1);
		ret.append('-').append(date.get(Calendar.DAY_OF_MONTH));
		if (timeWanted) {
			ret.append(' ').append(date.get(Calendar.HOUR_OF_DAY)).append(':');
			ret.append(date.get(Calendar.MINUTE)).append(':').append(
					date.get(Calendar.SECOND));
		}
		return ret.toString();
	}

	/**
	 * Log the excution results.
	 */
	public void log(String str) {
		calNow.setTimeInMillis(System.currentTimeMillis());
		System.out.println('[' + getDateString(calNow, true) + "] " + str);
	}

	/**
	 * Copy a file to another.
	 */
	public void copyFileToFile(File sourceFile, File destFile) throws Exception {
		InputStream in = new FileInputStream(sourceFile);
		OutputStream out = new FileOutputStream(destFile);
		byte[] buf = new byte[1024 * 16];
		int length;
		while ((length = in.read(buf)) > 0) {
			out.write(buf, 0, length);
		}
		in.close();
		out.flush();
		out.close();
	}

	/**
	 * Override the setup method.
	 * 
	 * @param context
	 *          context of this posEval object
	 */
	protected void setup(Context context) {
		loadProperties();
		lastRemodelTimeMilli = 0;
		lastEvalTimeMilli = 0;
		stateChangeTimes = new LinkedList();
	}

	/**
	 * This method overrides the map() method of class Mapper.
	 * 
	 * @param key
	 *          key of the data split
	 * @param value
	 *          contents of the data split, which is the station name
	 * @param context
	 *          context of this posEval object
	 * 
	 * @throws InterruptedException
	 * 
	 * @throws IOException
	 */
	protected void map(Object key, Text value, Context context)
			throws IOException, InterruptedException {
		siteName = value.toString();
		modelFilePrefix = siteName + "_POS_model";
		modelInputPrefix = siteName + "_input";
		siteModelDir = modelParentDir + File.separator + siteName + File.separator;
		siteEvalDir = evalParentDir + File.separator + siteName + File.separator;

		File fDir = new File(siteModelDir);
		if (!fDir.exists()) {
			fDir.mkdirs();
		}
		fDir = new File(siteEvalDir);
		if (!fDir.exists()) {
			fDir.mkdirs();
		}

		checkModelExists();
		lastModelUsableValue = false;

		run();
	}

	/***
	 * 
	 * Assistant classes for modeling.
	 * 
	 * Code from sensorgrid.sopac.gps.filters.RealtimeRdahmmRunner.java.
	 * 
	 */
	protected class ModelThread extends Thread {
		private int threadNum = 10;
		ThreadPool tPool;
		String trainShPath;
		String inputPath;
		double maxLValue = 0;

		public ModelThread(String trainShPath, String inputPath, ThreadPool tPool) {
			this.trainShPath = trainShPath;
			this.inputPath = inputPath;
			this.tPool = tPool;
		}

		public void run() {
			if (tPool != null)
				modelMultipleOneTries();
			else
				modelDirectTenTries();
		}

		private void modelDirectTenTries() {
			log("Starting to build model for " + inputPath);
			String res = exec(trainShPath + " " + inputPath + " 5 1234 10");
			if (res.length() == 0) {
				log("Model successfully built");
			} else {
				log("Results for building model for " + inputPath + ":");
				log(res);
			}
		}

		private void modelMultipleOneTries() {
			long st = System.currentTimeMillis();
			int idx = inputPath.lastIndexOf(File.separatorChar);
			String sourceDir = inputPath.substring(0, idx);
			String fileName = inputPath.substring(idx + 1);
			try {
				ModelOneTry[] oneTryList = new ModelOneTry[threadNum];
				Random random = new Random(1234);
				for (int i = 0; i < threadNum; i++) {
					String destDirStr = sourceDir + File.separatorChar + "tmp" + i;
					File destDir = new File(destDirStr);
					destDir.mkdirs();
					String destPath = destDir + File.separator + fileName;
					copyFileToFile(new File(inputPath), new File(destPath));
					oneTryList[i] = new ModelOneTry(trainShPath, destPath, random
							.nextInt(9999));
					tPool.assign(oneTryList[i]);
				}
				tPool.complete();

				maxLValue = oneTryList[0].getLValue();
				int idxMax = 0;
				for (int i = 1; i < threadNum; i++) {
					double v = oneTryList[i].getLValue();
					if (v > maxLValue) {
						maxLValue = v;
						idxMax = i;
					}
				}
				File tmpDir = new File(oneTryList[idxMax].getInputDir());
				File[] tmpFiles = tmpDir.listFiles();
				for (int i = 0; i < tmpFiles.length; i++) {
					String modelFilePath = sourceDir + File.separator
							+ tmpFiles[i].getName();
					copyFileToFile(tmpFiles[i], new File(modelFilePath));
				}

				for (int i = 0; i < oneTryList.length; i++) {
					tmpDir = new File(oneTryList[i].getInputDir());
					tmpFiles = tmpDir.listFiles();
					for (int j = 0; j < tmpFiles.length; j++)
						tmpFiles[j].delete();
					tmpDir.delete();
				}

				long et = System.currentTimeMillis();
				log("Starting time:" + st + "; ending time:" + et + "; interval:"
						+ (et - st));
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}

	protected class ModelOneTry implements Runnable {
		protected String inputPath;
		protected String trainShPath;
		protected int randomSeed;

		public ModelOneTry(String trainShPath, String inputPath, int randomSeed) {
			this.trainShPath = trainShPath;
			this.inputPath = inputPath;
			this.randomSeed = randomSeed;
		}

		public void run() {
			String res = exec(trainShPath + " " + inputPath + " 5 " + randomSeed
					+ " 1");
			if (res.length() == 0) {
				log("Model successfully built for seed " + randomSeed);
			} else {
				log("Results for building model for seed " + randomSeed + ":");
				log(res);
			}
		}

		public String getInputDir() {
			int idx = inputPath.lastIndexOf(File.separatorChar);
			if (idx < 0)
				return "";
			else
				return inputPath.substring(0, idx);
		}

		public double getLValue() {
			try {
				int idx = inputPath.lastIndexOf('.');
				String path = inputPath.substring(0, idx + 1) + "L";
				BufferedReader br = new BufferedReader(new FileReader(path));
				double ret = Double.parseDouble(br.readLine());
				br.close();
				return ret;
			} catch (Exception e) {
				e.printStackTrace();
				return 0;
			}
		}
	}

}